package com.jubull.rule.engine.entry;

import com.jubull.rule.engine.beans.EventBean;
import com.jubull.rule.engine.functions.RuleMatchKeyedProcessFunction;
import com.jubull.rule.engine.functions.json2LogBeanFunction;
import com.jubull.rule.engine.sources.KafkaSourceBuilder;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


import java.util.Objects;

/**
 * @author liuyun
 * @title: Main
 * @projectName jubull_marking
 * @description: 智能营销系统规则系统
 *                  需求：
 *                      获取用户事件，计算如下规则，输出结果
 *                      规则：
 *                          处罚事件：K事件，事件属性（p2 = v1）
 *                          画像属性条件：tog87 = v2 , tag26 -= v1
 *                          行为次数条件： 2021-06-18 ~ 当前，事件 C【p6 = v8, p12 = v5】做过 >= 2次
 * @date 2022-3-25 00:24
 */
public class Main {
    public static void main(String[] args) throws Exception {
        //构建env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        //读取kafka中的用户行为日志
        KafkaSourceBuilder kafkaSourceBuilder = new KafkaSourceBuilder();
        DataStreamSource<String> dss = env.addSource(kafkaSourceBuilder.build("jubull_applog"));
        //json解析
        SingleOutputStreamOperator<EventBean> dsBean = dss.map(new json2LogBeanFunction()).filter(Objects::nonNull);

        //keyBy: 按deviceId
        KeyedStream<EventBean, String> eventBeanStringKeyedStream = dsBean.keyBy(EventBean::getDeviceId);
        //规则计算
        SingleOutputStreamOperator<Object> dsMatchResult = eventBeanStringKeyedStream.process(new RuleMatchKeyedProcessFunction());


        env.execute();
    }
}
